package spark.yk.java.maven;
import org.apache.spark.sql.SparkSession;

import java.util.List;

import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.hive.*;

public class SparkSQLDemo {

	public static void main(String[] args) {
		
		
		
		/**
         * 第1步：创建Spark的配置对象SparkConf，设置Spark程序的运行时的配置信息，
         * 例如说通过setMaster来设置程序要链接的Spark集群的Master的URL,如果设置
         * 为local，则代表Spark程序在本地运行，特别适合于机器配置条件非常差（例如 只有1G的内存）的初学者 *
         */
		String uri = "hdfs://192.168.1.185:9000";  
	    String dir = "/user/output1";  
	    String parentDir = "/user";  
	    String remoteFile = uri+dir + "/README.md";
		// TODO Auto-generated method stub
		String logFile = remoteFile; // Should be some file on your system
		SparkConf conf = new SparkConf().setAppName("Simple Application").setMaster("spark://192.168.1.185:7077");

		// TODO Auto-generated method stub
		// office web spark sql and dataframe guide
		// http://spark.apache.org/docs/latest/sql-programming-guide.html
		SparkSession spark = SparkSession
				  .builder()
				  .appName("Java Spark SQL basic example")
				  .config(conf)
				  .enableHiveSupport()
				  .getOrCreate();
		
		
		/**
         * 第2步：创建SparkContext对象
         * SparkContext是Spark程序所有功能的唯一入口，无论是采用Scala、Java、Python
         * 、R等都必须有一个SparkContext(不同的语言具体的类名称不同，如果是Java的话则为JavaSparkContext)
         * SparkContext核心作用：初始化Spark应用程序运行所需要的核心组件，包括DAGScheduler、TaskScheduler、
         * SchedulerBackend 同时还会负责Spark程序往Master注册程序等
         * SparkContext是整个Spark应用程序中最为至关重要的一个对象
         */
		@SuppressWarnings("resource")  
        //JavaSparkContext sc = new JavaSparkContext(conf);  
		//sc.addJar("D:\\Java\\workspace\\SparkJavaDemo\\SparkJavaDemo.jar");
		//HiveContext hivecontext = new HiveContext(sc.sc());
	    
		
		
		/**
         * 第3步：根据具体的数据来源（HDFS、HBase、Local FS、DB、S3等）通过JavaSparkContext来创建JavaRDD
         * JavaRDD的创建基本有三种方式：根据外部的数据来源（例如HDFS）、根据Scala集合、由其它的RDD操作
         * 数据会被RDD划分成为一系列的Partitions，分配到每个Partition的数据属于一个Task的处理范畴
         * 注意：文件路径不能直接用Windows路径中的反斜扛\，要改成Linux下的斜扛/
         */
        //JavaRDD<String> lines = sc.textFile(logFile);
		
		Dataset<Row> df = spark.sql("select * from MysqlTest");

		// Displays the content of the DataFrame to stdout
		df.show();
		
		printLines(df);
		
		
		 spark.stop();
	}
	
	private static void printLines(Dataset<Row> df){
		
		List<Row> output = df.collectAsList();
		
		for (Row row : output) {  
            String lines =  row.mkString(); 
            System.out.println(lines);
          }  
		
	}

}
